# JoinTableRootJoinNode.java

    package org.codefilarete.stalactite.engine.runtime.load;

    import javax.annotation.Nullable;
    import java.util.Set;
    import java.util.function.BiConsumer;
    import java.util.function.Function;
    import java.util.stream.Collectors;

    import org.codefilarete.stalactite.engine.runtime.ConfiguredRelationalPersister;
    import org.codefilarete.stalactite.engine.runtime.load.EntityInflater.EntityMappingAdapter;
    import org.codefilarete.stalactite.engine.runtime.load.EntityTreeInflater.TreeInflationContext;
    import org.codefilarete.stalactite.engine.runtime.load.JoinRowConsumer.ExcludingJoinRowConsumer;
    import org.codefilarete.stalactite.engine.runtime.load.JoinRowConsumer.RootJoinRowConsumer;
    import org.codefilarete.stalactite.engine.runtime.load.MergeJoinNode.MergeJoinRowConsumer;
    import org.codefilarete.stalactite.mapping.EntityMapping;
    import org.codefilarete.stalactite.mapping.RowTransformer;
    import org.codefilarete.stalactite.query.api.Selectable;
    import org.codefilarete.stalactite.sql.ddl.structure.Column;
    import org.codefilarete.stalactite.sql.ddl.structure.Table;
    import org.codefilarete.stalactite.sql.result.ColumnedRow;
    import org.codefilarete.tool.Duo;
    import org.codefilarete.tool.Reflections;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;

    /**
     * Particular {@link JoinRoot} made to handle join-table polymorphic case : polymorphic entity instantiation is the core focus of it.
     * Here are the hotspots: identifier is given by the subclass which find its id in the row (see {@link JoinTablePolymorphicJoinRootRowConsumer#findSubInflater(ColumnedRow)}),
     * and while doing it, it remembers which consumer made it. Then while instantiating entity we invoke it to get right entity type (parent mapping
     * would only give parent entity, which, out of being a wrong approach, can be an abstract type). Then instance is filled up with parent properties
     * by calling merging method (see line 98).
     * Finally, {@link JoinTablePolymorphicJoinRootRowConsumer} must extend {@link ExcludingJoinRowConsumer} to give next branch to be consumed by
     * {@link EntityTreeInflater} to avoid "dead" branch to be read : we give it according to the consumer which found the identifier.
     *
     * @author Guillaume Mary
     */
    public class JoinTableRootJoinNode<C, I, T extends Table<T>> extends JoinRoot<C, I, T> {

    	private final Set<? extends ConfiguredRelationalPersister<C, I>> subPersisters;
    	private final Set<Column<T, ?>> selectableColumns;
    	private JoinTablePolymorphicJoinRootRowConsumer<C, I> rootConsumer;

    	public JoinTableRootJoinNode(EntityJoinTree<C, I> tree,
    								 ConfiguredRelationalPersister<C, I> mainPersister,
    								 Set<? extends ConfiguredRelationalPersister<C, I>> subPersisters,
    								 Set<? extends Column<T, ?>> selectableColumns,
    								 T mainTable) {
    		super(tree, new EntityMappingAdapter<>(mainPersister.<T>getMapping()), mainTable);
    		this.subPersisters = subPersisters;
    		this.selectableColumns = (Set<Column<T, ?>>) selectableColumns;
    	}

    	@Override
    	public Set<Selectable<?>> getColumnsToSelect() {
    		return (Set) this.selectableColumns;
    	}

    	@Override
    	public RootJoinRowConsumer<C, I> toConsumer(JoinNode<C, T> joinNode) {
    		RowTransformer<C> rootRowTransformer = getEntityInflater().getRowTransformer();
    		Set<SubPersisterConsumer<C, I>> subEntityConsumer = subPersisters.stream().map(subPersister -> {
    			EntityMapping<C, I, T> mapping = subPersister.getMapping();
    			return new SubPersisterConsumer<>(
    					row -> mapping.getIdMapping().getIdentifierAssembler().assemble(row),
    					mapping.getClassToPersist(),
    					mapping.getRowTransformer());
    		}).collect(Collectors.toSet());
    		rootConsumer = new JoinTablePolymorphicJoinRootRowConsumer<>(joinNode, rootRowTransformer, subEntityConsumer,
    				getConsumptionListener() == null ? null : (rootEntity, row) -> getConsumptionListener().onNodeConsumption(rootEntity, EntityTreeInflater.currentContext().getDecoder(joinNode)));
    		return rootConsumer;
    	}

    	public void addSubPersister(ConfiguredRelationalPersister<C, I> persister, MergeJoinRowConsumer<C> subConsumer) {
    		rootConsumer.subConsumers.forEach(pawnConsumer -> {
    			if (pawnConsumer.subEntityType == persister.getClassToPersist()) {
    				pawnConsumer.subPropertiesApplier = subConsumer;
    				pawnConsumer.identifierAssembler = row -> persister.getMapping().getIdMapping().getIdentifierAssembler().assemble(row);
    			}
    		});
    	}

    	static class SubPersisterConsumer<C, I> {
    		private Function<ColumnedRow, I> identifierAssembler;
    		private final Class<C> subEntityType;
    		private final RowTransformer<C> subEntityFactory;
    		private MergeJoinRowConsumer<C> subPropertiesApplier;

    		private SubPersisterConsumer(Function<ColumnedRow, I> identifierAssembler,
    									 Class<C> subEntityType,
    									 RowTransformer<C> subEntityFactory) {
    			this.identifierAssembler = identifierAssembler;
    			this.subEntityType = subEntityType;
    			this.subEntityFactory = subEntityFactory;
    		}
    	}

    	static class JoinTablePolymorphicJoinRootRowConsumer<C, I> implements ExcludingJoinRowConsumer<C, I> {

    		private static final ThreadLocal<MergeJoinRowConsumer<?>> CURRENTLY_FOUND_CONSUMER = new ThreadLocal<>();

    		protected final Logger LOGGER = LoggerFactory.getLogger(this.getClass());

    		private final JoinNode<C, ?> node;
    		private final RowTransformer<C> rootRowTransformer;
    		private final Set<SubPersisterConsumer<C, I>> subConsumers;

    		/** Optional listener of ResultSet decoding */
    		@Nullable
    		private final BiConsumer<C, ColumnedRow> consumptionListener;

    		private JoinTablePolymorphicJoinRootRowConsumer(JoinNode<C, ?> node,
    														RowTransformer<C> rootRowTransformer,
    														Set<SubPersisterConsumer<C, I>> subConsumers,
    														@Nullable BiConsumer<C, ColumnedRow> consumptionListener) {
    			this.node = node;
    			this.rootRowTransformer = rootRowTransformer;
    			this.subConsumers = subConsumers;
    			this.consumptionListener = consumptionListener;
    		}

    		@Override
    		public JoinNode<C, ?> getNode() {
    			return node;
    		}

    		@Override
    		public EntityReference<C, I> createRootInstance(ColumnedRow row, TreeInflationContext context) {
    			Duo<I, SubPersisterConsumer<C, I>> subInflater = findSubInflater(row);
    			EntityReference<C, I> result = null;
    			if (subInflater == null) {
    				CURRENTLY_FOUND_CONSUMER.remove();
    			} else {
    				CURRENTLY_FOUND_CONSUMER.set(subInflater.getRight().subPropertiesApplier);
    				C entity = context.giveEntityFromCache(subInflater.getRight().subEntityType, subInflater.getLeft(), () -> {
    					LOGGER.debug("Instantiating entity of type {}", subInflater.getRight().subEntityType);
    					ColumnedRow subInflaterRow = EntityTreeInflater.currentContext().getDecoder(subInflater.getRight().subPropertiesApplier.getNode());
    					// by using transform(..) we instantiate the right type (the sub-entity one) and fill it with sub-entity properties
    					return subInflater.getRight().subEntityFactory.transform(subInflaterRow);
    				});
    				// applying parent properties to the entity
    				rootRowTransformer.applyRowToBean(row, entity);
    				if (consumptionListener != null) {
    					consumptionListener.accept(entity, row);
    				}
    				result = new EntityReference<>(entity, subInflater.getLeft());
    			}
    			return result;
    		}

    		@Nullable
    		/* Optimized, from 530 000 nanos to 65 000 nanos at 1st exec, from 40 000 nanos to 12 000 nanos on usual run */
    		public Duo<I, SubPersisterConsumer<C, I>> findSubInflater(ColumnedRow row) {
    			// @Optimized : use for & return instead of stream().map().filter(notNull).findFirst()
    			for (SubPersisterConsumer<C, I> pawn : subConsumers) {
    				ColumnedRow subInflaterRow = EntityTreeInflater.currentContext().getDecoder(pawn.subPropertiesApplier.getNode());
    				I assemble = pawn.identifierAssembler.apply(subInflaterRow);
    				if (assemble != null) {
    					return new Duo<>(assemble, pawn);
    				}
    			}
    			return null;
    		}

    		/**
    		 * Gives the row consumers that instantiates sub-entities but weren't detected in very previous identifier detection.
    		 * This allows to keep effective sub-entity merger and relation consumers of parent entity while iterating of joins during tree inflation.
    		 *
    		 * @return the consumers that shouldn't be taken into account in next tree iteration
    		 */
    		public Set<JoinRowConsumer> giveExcludedConsumers() {
    			return subConsumers.stream()
    					.map(subConsumer -> subConsumer.subPropertiesApplier)
    					.filter(consumerPawn -> CURRENTLY_FOUND_CONSUMER.get() != consumerPawn)
    					.collect(Collectors.toSet());
    		}

    		/**
    		 * Implemented for debug. DO NOT RELY ON IT for anything else.
    		 */
    		@Override
    		public String toString() {
    			return Reflections.toString(this.getClass());
    		}
    	}
    }